package com.etung.iot.demo.mqtt;


import com.etung.iot.demo.handler.HandlerFactory;
import com.etung.iot.demo.handler.TopicHandler;
import com.etung.iot.demo.utils.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.aop.support.AopUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.Optional;

@Service
@Slf4j
public class EmqxService {


    public void subscribe() {
        Map<String, Object> beansWithAnnotation = SpringContextUtils.getApplicationContext().getBeansWithAnnotation(MqttMessageListener.class);
        Optional.of(beansWithAnnotation).ifPresent((Map<String, Object> mqttListener) -> {
            beansWithAnnotation.forEach((name, bean) -> {
                Class<?> clazz = AopUtils.getTargetClass(bean);
                MqttMessageListener annotation = clazz.getAnnotation(MqttMessageListener.class);
                String[] topics = annotation.topic();
                if (!StringUtils.isEmpty(topics)) {
                    for (String topic : topics) {
                        try {
                            EmqxClient.client.subscribe(topic, annotation.qos().value());
                        } catch (MqttException e) {
                            log.error("======>>>>>订阅主题失败 error={}", e.getMessage());
                            e.printStackTrace();
                        }
                    }
                }
            });

        });
    }

    /**
     * 消息回调方法
     *
     * @param topic       主题
     * @param mqttMessage 消息体
     */
    @Async("threadPoolTaskExecutor")
    public void subscribeCallback(String topic, MqttMessage mqttMessage) throws InterruptedException {

        /**测试线程池使用*/
//        log.info("====>>>>线程名--{}", Thread.currentThread().getName());
        /**模拟耗时操作*/
        // Thread.sleep(1000);
        // subscribe后得到的消息会执行到这里面
        String message = new String(mqttMessage.getPayload());
//        log.info("接收消息主题 : " + topic);
//        log.info("接收消息Qos : " + mqttMessage.getQos());
//        log.info("接收消息内容 : " + message);

        TopicHandler topicHandler = HandlerFactory.getTopicHandler(topic);
        if (topicHandler != null) {
            try{
                topicHandler.handler(topic, message);
            }catch (Exception e){
                log.error(e.getMessage());
            }
        }
    }



}
